package io.jjz.dao;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.coprocessor.AggregationClient;
import org.apache.hadoop.hbase.client.coprocessor.LongColumnInterpreter;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Repository;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;

@Repository
public class CrudDao  { //extends HBaseDao
    private static final Logger LOGGER = LoggerFactory.getLogger(CrudDao.class);

    @Resource(name = "hBaseConnection")
    private Connection connection;

    private ExecutorService workerPool;

    @PostConstruct
    public void init() {
        workerPool = Executors.newFixedThreadPool(4);
    }

    @PreDestroy
    public void destroy() {
        workerPool.shutdown();
    }

    /**
     * 添加或者更新某个cell数据
     */
    public boolean putRow(String tableName, byte[] rowKey, String family, String column, byte[] value) {
        // Table是个很轻量的对象，可以为每个请求创建一个Table， 2.0以后版本实现为线程安全的类
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            Put put = new Put(rowKey);
            put.addColumn(Bytes.toBytes(family), Bytes.toBytes(column), value);
            table.put(put);
            return true;
        } catch (IOException e) {
            return false;
        }
    }

    /**
     * 添加或更新同一列族下多个列
     */
    public boolean putRows(String tableName, byte[] rowKey, String family, List<Pair<String, byte[]>> pairs) {
        final List<Put> puts = transfer(rowKey, family, pairs);
        return putSync(tableName, puts);
    }

    public boolean putRows(String tableName, byte[] rowKey, String family, Map<String, byte[]> value) {
        final List<Put> puts = transfer(rowKey, family, value);
        return putSync(tableName, puts);
    }

    private boolean putSync(String tableName, List<Put> puts) {
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            table.put(puts);
            return true;
        } catch (IOException e) {
            return false;
        }
    }

    // 异步批量
    public Future putRowsAsync(String tableName, byte[] rowKey, String family, List<Pair<String, byte[]>> pairs) {
        final List<Put> puts = transfer(rowKey, family, pairs);
        return putAsync(tableName, puts);
    }

    public Future putRowsAsync(String tableName, byte[] rowKey, String family, Map<String, byte[]> value) {
        final List<Put> puts = transfer(rowKey, family, value);
        return putAsync(tableName, puts);
    }

    private Future<Void> putAsync(String tableName, List<Put> puts) {
        final BufferedMutator.ExceptionListener listener = (e, mutator) -> {
            for (int i = 0; i < e.getNumExceptions(); i++) {
                LOGGER.warn("Failed to sent put " + e.getRow(i));
            }
        };

        BufferedMutatorParams params = new BufferedMutatorParams(TableName.valueOf(tableName)).listener(listener);
        try (BufferedMutator mutator = connection.getBufferedMutator(params)) {
            return workerPool.submit(() -> {
                mutator.mutate(puts);
                mutator.flush();
                return null;
            });

        } catch (IOException e) {
            throw new RuntimeException(e.getMessage());
        }
    }

    private List<Put> transfer(byte[] rowKey, String family, List<Pair<String, byte[]>> pairs) {
        final ArrayList<Put> puts = new ArrayList<>();
        for (Pair pair : pairs) {
            Put put = new Put(rowKey);
            put.addColumn(Bytes.toBytes(family), ((String) pair.getFirst()).getBytes(), (byte[]) pair.getSecond());
            puts.add(put);
        }
        return puts;
    }

    private List<Put> transfer(byte[] rowKey, String family, Map<String, byte[]> map) {
        final ArrayList<Put> puts = new ArrayList<>();
        for (Map.Entry<String, byte[]> entry : map.entrySet()) {
            Put put = new Put(rowKey);
            put.addColumn(Bytes.toBytes(family), Bytes.toBytes(entry.getKey()), entry.getValue());
            puts.add(put);
        }
        return puts;
    }


    // 查询一个cell的数据
    public byte[] getCell(String tableName, byte[] rowKey, String family, String column) {
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            Get get = new Get(rowKey);
            get.addColumn(family.getBytes(), column.getBytes());
            Result result = table.get(get);
            return result.getValue(family.getBytes(), column.getBytes());
        } catch (IOException e) {
            throw new RuntimeException(e.getMessage());
        }
    }


    // 查询某行的一个列族的数据
    public Map<byte[], byte[]> getCells(String tableName, byte[] rowKey, String family) {
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            Get get = new Get(rowKey);
            get.addFamily(family.getBytes());
            Result result = table.get(get);
            return result.getFamilyMap(family.getBytes());
        } catch (IOException e) {
            throw new RuntimeException(e.getMessage());
        }
    }

    // 查询一行数据
    public List<Cell> getRow(String tableName, byte[] rowKey) {
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            final Get get = new Get(rowKey);
            final Result result = table.get(get);
            return result.listCells();
        } catch (IOException e) {
            throw new RuntimeException(e.getMessage());
        }
    }

    // 一次查询多个rowkey
    public Result[] getMultiRow(String tableName, List<byte[]> rowKeys) {
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            List<Get> getList = new ArrayList<>();
            for (byte[] row : rowKeys) {
                getList.add(new Get(row));
            }
            return table.get(getList);
        } catch (IOException e) {
            throw new RuntimeException(e.getMessage());
        }
    }


    /**
     * 删除指定rowKey那行数据
     */
    public boolean deleteRow(String tableName, byte[] rowKey) {
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            Delete delete = new Delete(rowKey);
            table.delete(delete);
            return true;
        } catch (IOException e) {
            return false;
        }
    }

    /**
     * 删除指定rowKey的某一列的数据
     */
    public boolean deleteCell(String tableName, byte[] rowKey, String family, String column) {
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            Delete delete = new Delete(rowKey);
            delete.addColumn(Bytes.toBytes(family), Bytes.toBytes(column));
            table.delete(delete);
            return true;
        } catch (IOException e) {
            return false;
        }
    }


    /**
     * 范围查找某一列, 包头不包尾
     */
    public List<byte[]> scanCells(String tableName, byte[] startRow, byte[] endRow,
                                      String familyName, String columnName, int limit) {
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            Scan scan = new Scan().withStartRow(startRow).withStopRow(endRow, false);
            if (limit > 0) {
                scan.setLimit(limit);
            }
            scan.addColumn(familyName.getBytes(), columnName.getBytes());
            try (ResultScanner scanner = table.getScanner(scan)) {
                final ArrayList<byte[]> cells = new ArrayList<>();
                for (Result result : scanner) {
                    final byte[] value = result.getValue(familyName.getBytes(), columnName.getBytes());
                    cells.add(value);
                }
                return cells;
            }
        } catch (IOException e) {
            throw new RuntimeException(e.getMessage());
        }
    }


    public List<Cell> scanColumn(String tableName, byte[] startRow, byte[] endRow, String family) {
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            Scan scan = new Scan().withStartRow(startRow).withStopRow(endRow, false);
            scan.setLimit(10);

            // todo 区别
            scan.addFamily(family.getBytes());
            //scan.setFilter(new FamilyFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(family))));

            try (ResultScanner scanner = table.getScanner(scan)) {
                final ArrayList<Cell> cells = new ArrayList<>();
                for (Result result : scanner) {
                    cells.addAll(result.listCells());
                }
                return cells;
            }
        } catch (IOException e) {
            throw new RuntimeException(e.getMessage());
        }
    }

    /**
     * 根据前缀匹配查询
     */
    public List<Cell> scanPrefixRowkey(String tableName, String prefix, int limit) {
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            Scan scan = new Scan();
            if (limit > 0) {
                scan.setLimit(limit);
            }

            // 相比直接使用filter更高效
            // 比如扫描 “def”开头的前缀，内部进行展开，设为扫描startRow为def，endRow为deg
            scan.setRowPrefixFilter(Bytes.toBytes(prefix));
            //scan.setFilter(new PrefixFilter(prefix.getBytes()));

            try (ResultScanner rs = table.getScanner(scan)) {
                ArrayList<Cell> list = new ArrayList<>();
                for (Result r : rs) {
                    list.addAll(r.listCells());
                }
                return list;
            }
        } catch (IOException e) {
            throw new RuntimeException(e.getMessage());
        }
    }

    /**
     * 统计rowKey范围的数量
     */
    public long countRow(String tableName, byte[] startRow, byte[] endRow) {
        AggregationClient aggregationClient = new AggregationClient(connection.getConfiguration());
        Scan scan = new Scan().withStartRow(startRow).withStopRow(endRow, false);
        try {
            return aggregationClient.rowCount(TableName.valueOf(tableName), new LongColumnInterpreter(), scan);
        } catch (Throwable throwable) {
            throw new RuntimeException(throwable.getMessage());
        }
    }


    // 统计含有某列的行数
    public long countColumn(String tableName, String familyName, String columnName, String value) throws Throwable {
        SingleColumnValueFilter columnValueFilter = new SingleColumnValueFilter(
                Bytes.toBytes(familyName), Bytes.toBytes(columnName),
                CompareFilter.CompareOp.EQUAL, Bytes.toBytes(value));  //new SubstringComparator(value)
        // 为true时不包含该列的行不会返回
        columnValueFilter.setFilterIfMissing(true);
        Scan scan = new Scan();
        scan.setFilter(columnValueFilter);

        AggregationClient aggregationClient = new AggregationClient(connection.getConfiguration());
        return aggregationClient.rowCount(TableName.valueOf(tableName), new LongColumnInterpreter(), scan);
    }

}